package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo7WindowProcess {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    //划分窗口
    val windowDS: WindowedStream[(String, Int), String, TimeWindow] = keyByDS
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

    /**
     * process: 在窗口之后使用process，可以获取到窗口内所有的数据
     */
    val countDS: DataStream[(String, Int)] = windowDS
      .process(new ProcessWindowFunction[(String, Int), (String, Int), String, TimeWindow] {

        /**
         * process:每个key每一个窗口执行一次
         *
         * @param key      ：单词
         * @param context  ：上下文对象，可以获取到窗口的时间
         * @param elements ：窗口内所有的数据
         * @param out      ：英语将数据发送到下游
         */
        override def process(key: String,
                             context: Context,
                             elements: Iterable[(String, Int)],
                             out: Collector[(String, Int)]): Unit = {
          //可以获取窗口的时间
          val endTime: Long = context.window.getEnd

          var count = 0
          for (elem <- elements) {
            //计算单词的数量
            count += 1
          }

          //将统计结果发送到下游
          out.collect((key, count))
        }
      })

    countDS.print()

    env.execute()
  }

}
